package com.practice.car.cardataapp.schelduer.spark.task.complaints;

import com.alibaba.fastjson.JSON;
import com.practice.car.cardataapp.entity.Complaints;
import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.ComplaintsRunner;
import lombok.Data;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import scala.Serializable;
import scala.Tuple2;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import static com.practice.car.cardataapp.schelduer.spark.Schedule.getConnection;

/**
 * demo-study project
 *
 * @author FlagDied
 * @date 2020/8/27
 */
public class ComplaintsRddApplicaiton implements Serializable, ComplaintsRunner {
	private Connection connection = null;

	@Override
	public void run() {
		String path =Schedule.complaintsHdfsUrl;//local | hdfs://localhost:9000
		//开启 SparkContext
		JavaSparkContext sc = new JavaSparkContext("local[2]", "rdd_application");
		JavaRDD<Complaints> complaintsJavaRDD = sc.parallelize(readJsonArray(path, Complaints.class,""));
		restrieveComplaints(complaintsJavaRDD);
		sortComplaints(complaintsJavaRDD);
		sc.close();
	}

	public void sortComplaints(JavaRDD<Complaints> rdd) {//投诉排行
		List list=null; //接收rdd
		List<String> list2=new ArrayList<String>();//去除无用符
		//complaints->(name,brand),n
		JavaPairRDD<Tuple2, Integer> prdd = rdd.mapToPair(complaints -> new Tuple2<>(new Tuple2(complaints.getName(),
				complaints.getBrand()),1)).foldByKey(2, new Function2<Integer, Integer, Integer>() {
			@Override
			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1+v2;
			}
		});
		list= Arrays.asList(prdd.collect().toArray());//接收rdd数据
		for (int i = 0; i < list.size(); i++) {
			String a=list.get(i).toString().replaceAll("\\(|\\)","");
			list.set(i,a);
			list2.add(list.get(i).toString());
		}
		//保存数据库
		try {
			Connection con =getConnection();
			System.out.println(list.size());
			PreparedStatement psql =con.prepareStatement("DELETE FROM `complaint_ranking`");
			psql.executeUpdate();
			for (int i = 0; i < list2.size(); i++) {
				String str[]=list2.get(i).split(",");
				psql = con.prepareStatement("insert into complaint_ranking(id,brand,name,amount) values(?,?,?,?) ");
				psql.setInt(1,i+1);
				psql.setString(2, str[1]);
				psql.setString(3, str[0]);
				psql.setInt(4, Integer.parseInt(str[2]));
				psql.executeUpdate();
			}
			con.close();
		} catch (SQLException throwables) {
			throwables.printStackTrace();
		}
	}

	public void restrieveComplaints(JavaRDD<Complaints> rdd) {//投诉查询
		List list=null;//接收rdd数据
		List<String> list2=new ArrayList<String>();//去除无用符
		//complaints->(brand,complaints),n
		JavaPairRDD<Tuple2, Integer> prdd = rdd.mapToPair(complaints -> new Tuple2<>(new Tuple2(complaints.getBrand(),
				complaints.getComplaints()),1)).foldByKey(2, new Function2<Integer, Integer, Integer>() {
			@Override
			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1+v2;
			}
		});
		list= Arrays.asList(prdd.collect().toArray()); //接收rdd
		for (int i = 0; i < list.size(); i++) {
			String a=list.get(i).toString().replaceAll("\\(|\\)|:|：","");
			list.set(i,a);
			list2.add(list.get(i).toString());
		}
		//保存
		try {
			Connection con =getConnection();
			System.out.println(list.size());
			PreparedStatement psql =con.prepareStatement("DELETE FROM `complaint`");
			psql.executeUpdate();
			for (int i = 0; i < list2.size(); i++) {
				String str[]=list2.get(i).split(",");
				psql = con.prepareStatement("insert into complaint(id,brand,complaint_type,amount) values(?,?,?,?) ");
				psql.setInt(1,i+1);
				psql.setString(2, str[0]);
				psql.setString(3, str[1]);
				psql.setInt(4, Integer.parseInt(str[2]));
				psql.executeUpdate();
			}
			con.close();
		} catch (SQLException throwables) {
			throwables.printStackTrace();
		} 

	}

	/*
	public Connection getcon()throws SQLException, ClassNotFoundException{
		Connection con = null;
		String driver = "com.mysql.cj.jdbc.Driver";
		String url = "jdbc:mysql://localhost/bigdatadb?serverTimezone=GMT";
		String user = "root";
		String password = "ming3734223";
		Class.forName(driver);
		con = DriverManager.getConnection(url,user,password);
		return con;
	}*/

	private <T> List<T> readJsonArray(String path, Class<T> clasz, String model) {//local模式，加载本地项目路径下的json文件，默认加载的是hdfs上的文件
		List<T> list = null;
		int available = 0;
		BufferedInputStream inputStream = null ;
		if ("local".equals(model)) {
			try {
				inputStream = new BufferedInputStream(new FileInputStream(path));
			} catch (FileNotFoundException e) {
				e.printStackTrace();
			}
		}else{
			Path hdfsPath = new Path(path);
			Configuration configuration = new Configuration();
			try {
				FileSystem fileSystem = new Path(path).getFileSystem(configuration);
				inputStream= new BufferedInputStream(fileSystem.open(hdfsPath));
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		try {
			available = inputStream.available();
			byte[] buffer = new byte[available];
			inputStream.read(buffer);
			inputStream.close();
			list = JSON.parseArray(new String(buffer, StandardCharsets.UTF_8), clasz);
		} catch (IOException e) {
			e.printStackTrace();
		}
		//将 text文件里的 json数据读取 List<Car>
		return list;
	}

	@Data
	static class complaint_ranking implements Serializable{
		String brand;
		String name;
		int amount;

		complaint_ranking(){
			brand="";
			name="";
			amount=0;
		}
	}


}
